package join

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object BasedWindowDoubleStream {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val stream1: DataStream[(Int, Int, Int)] = env.fromElements(
      (1, 11, 1000),
      (1, 12, 1500),
      (1, 13, 2000),
      (1, 14, 2500),
      (1, 15, 3000),
      (1, 16, 3500),
      (2, 17, 2500),
      (2, 18, 3000),
      (2, 19, 3500)
    ).assignAscendingTimestamps(x => x._3)

    val stream2: DataStream[(Int, Int, Int)] = env.fromElements(
      (1, 21, 1000),
      (1, 22, 1500),
      (1, 23, 2000),
      (1, 24, 2500),
      (1, 25, 3000),
      (1, 26, 3500),
      (2, 27, 2500),
      (2, 28, 3000),
      (2, 29, 3500)
    ).assignAscendingTimestamps(x => x._3)

    stream1.join(stream2)
      .where(x => x._1)
      .equalTo(x => x._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(1)))
      .apply((stream1, stream2) => {println(stream1._2 + "-" +stream2._2)})
      .print()

    env.execute()
  }
}
